import dao.CountInfoMapper;
import model.CountInfo;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import utils.CountInfoDao;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;

public class WriteMysql {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("spark_wordcount_java");

        JavaSparkContext sc = new JavaSparkContext(conf);


        JavaRDD<String> lines = sc.textFile("src/main/resources/data");
        JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
        JavaPairRDD<String, Integer> pairRDD = words.mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> result = pairRDD.reduceByKey((v1, v2) -> v1 + v2);
        //result.foreach(tuple2 -> System.out.println(tuple2._1 + ":" + tuple2._2));

        /*result.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
            @Override
            public void call(Iterator<Tuple2<String, Integer>> partitionRecords) throws Exception {


                while (partitionRecords.hasNext()) {
                    CountInfoMapper countInfoMapper = CountInfoDao.getInstance();
                    Tuple2<String, Integer> nex = partitionRecords.next();
                    CountInfo countInfo = new CountInfo(nex._1, Long.parseLong(nex._2 + ""));
                    System.out.println(nex._1 + ":" + nex._2);
                    countInfoMapper.insert(countInfo);
                }

            }
        });*/
        System.out.println("par:" + result.getNumPartitions());

        result.foreachPartition(it -> {
            long startTime= System.currentTimeMillis();
            AtomicInteger count = new AtomicInteger(0);
            CountInfoMapper countInfoMapper = CountInfoDao.getInstance();
            it.forEachRemaining(v -> {
                        CountInfo countInfo = new CountInfo(v._1, Long.parseLong(v._2 + ""));
                        count.getAndIncrement();
                        qpsControl(count, startTime);
                        countInfoMapper.insert(countInfo);
                    }
            );
        });
        sc.stop();
    }

    //qps限流
    private static void qpsControl(AtomicInteger count, long time) {
        int actuarQps = (int) (count.get() / ((System.currentTimeMillis() - time) / 1000));
        if (actuarQps > 1000) {
            try {
                Thread.sleep(3000);//限流
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } else {
            //todo
            return;
        }
    }
}
